package com.o2o.cleaning.month.platform.ebusiness_plat.miya

import com.alibaba.fastjson.JSON
import com.mongodb.spark.MongoSpark
import com.o2o.cleaning.month.platform.ebusiness_plat.brand_modular.brand_join_res
import org.apache.spark.SparkContext
import org.apache.spark.sql.functions.lit
import org.apache.spark.sql.{DataFrame, SparkSession}

object MiYa_618 {
  var platform = "miya"           //平台名称
  var year = "2021"               //当月的年份、月份
  var month = "6"
  var lastMonth = "5"             //上月的月份 （销量相减时用到）
//  var timeStamp = TimesYearAll.TIME202005  //每个月固定时间戳
  var timeStamp = "1623945600"  //每个月固定时间戳
  //mongo库中的库名称
  var database = "Miya"
  //  mongo库中的集合名称
  var collection = "miya_2106"
  var last_collection = "miya_2105"
//Miya.miya_2006
  //原始数据
//  var sourcePath = s"s3a://o2o-sourcedata-2021/obs-source-2021/620/${platform}/"
  var sourcePath = s"s3a://o2o-sourcedata-2021/obs-source-2021/615/${platform}/"
//  var sourcePath = s"s3a://o2o-sourcedata/obs-source-2020/${month}/${platform}/${collection}"
  //上月数据路径
  var lastMonthPath = s"s3a://o2o-sourcedata/obs-source-2021/${lastMonth}/${platform}/${last_collection}/"
  //三级id分类路径
  var subPath = "s3a://o2o-dimension-table/category_table/cate/cate0401/miya/subCategoryId/*"
  //商品的清洗结果路径
//  var resultPath = s"s3a://o2o-dataproces-group/zyf/2021/${month}/${platform}/good"
//  var resultPath = s"s3a://o2o-dataproces-group/zyf/2021/${month}20/${platform}/good"
  var resultPath = s"s3a://o2o-dataproces-group/zyf/2021/${month}18/${platform}/good"
  //提取的店铺路径
  var shopPath = s"s3a://o2o-dataproces-group/zyf/2021/${month}/${platform}/shop"
  //new新增数据
//  var newPath = s"s3a://o2o-dataproces-group/zyf/2021/${month}/${platform}/newAdd_test"
  var newPath = s"s3a://o2o-dataproces-group/zyf/2021/${month}/${platform}/newAdd"
  //校验路径
  var strPath = "s3a://o2o-dataproces-group/zyf/"

  //    老品牌表路径
//  val brandTable_old = "s3a://o2o-dataproces-group/zyf/2021/618/miya_620/"
  //    品牌表结果路径
  val brandTable_Result = "s3a://o2o-dataproces-group/zyf/2021/618/miya_6_20/"
//      新增品牌路径
  val brandTable_addNew = "s3a://o2o-dataproces-group/zyf/newAddBrand/2021/618/miya_6_20/"
//    老品牌表路径
val brandTable_old = s"s3a://o2o-dimension-table/brandName_table/${year}/${month}/${platform}/"
  //    品牌表结果路径
//  val brandTable_Result = s"s3a://o2o-dimension-table/brandName_table/${year}/${month.toInt+1}/${platform}/"
  //    新增品牌路径
//  val brandTable_addNew = s"s3a://o2o-dimension-table/brandName_table/newAddBrand/${year}/${month}/${platform}/"

  def main(args: Array[String]): Unit = {
//   蜜芽	韩汛	日采	mongodb(149)	Miya.miya_2006	凑18天销量
    //spark mongo连接配置
    val spark = SparkSession.builder()
      .master("local[*]")
      .config("spark.debug.maxToStringFields", "10000")
      .appName("MongoSparkConnectorIntro")
      .config("spark.mongodb.input.uri", "mongodb://root:O2Odata123!@ 192.168.0.149:27017/admin")
      .config("spark.mongodb.input.database", s"${database}")
      .config("spark.mongodb.input.collection", s"${collection}")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .getOrCreate()
    //obs设置
    var sc: SparkContext = spark.sparkContext
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")
    sc.setLogLevel("ERROR")

    //备份MongoDB数据
//    mongoExport(spark,sc,month, platform, database, collection)
    //清洗计算
    var frame2: DataFrame = miyaCaculate(spark, s"${sourcePath}/*",lastMonthPath)
      .where("sellCount>0").where("priceText>0").drop("add_to_field")
    frame2.registerTempTable("t2")

    //打标签
    var s_002 = frame2
      .withColumn("shopId", lit("11111111")).withColumn("timeStamp", lit(s"${timeStamp}"))
      .withColumn("shopName", lit("蜜芽自营店"))
      .withColumn("platformName", lit("蜜芽")).withColumn("platformId", lit("11")).withColumn("shopType", lit("B"))
      .where("sellCount>0").where("priceText>0")

    //关联分类
    val frame = miyaCate(spark, s_002)
    //关联地址
    val frame1 = miyaAddress(spark, frame)
    //关联品牌
    val brand = new brand_join_res
    //读取老品牌表路径
    brand.brandJoinResult(frame1,resultPath,year.toInt,month.toInt,platform,spark)
    //商品数据落地
//    frame1.repartition(1).write.json(resultPath)
    //校验数据---落地到mysql和保存csv及json文件
    //checkData(spark,frame1)

  }

  /**
    * 从mongo中拉取数据到OBS
    *
    * @param month
    * @param platform
    * @param collection
    */
  def mongoExport(spark:SparkSession,sc:SparkContext,month: String, platform: String, database: String, collection: String): Unit = {

//    MongoSpark.load(sc).foreachPartition(partition=>{
//      partition.map(line=>{
//
//      })
//    })

    val values = MongoSpark.load(sc).map(line => {
      val nObject = JSON.parseObject(line.toJson())

      //修改蜜芽是否为境外商品的依据
      var taxFreeZone  = nObject.get("taxFreeZone").toString
      if (taxFreeZone.equals("-1")){
        nObject.put("is_oversea","false")
      }

      if (!taxFreeZone.equals("-1")){
        nObject.put("is_oversea","true")
      }

      //所有字段转为string
      val keys = nObject.keySet().iterator()
      while (keys.hasNext){
        var key = keys.next()
        nObject.put(key,nObject.get(key).toString)
      }

      nObject.remove("_id")
      nObject.toString
    })
    spark.read.json(values).repartition(1).write.orc(sourcePath)
  }

  /** *
    * 商品计算一：根据commentCount计算
    *   关联上的数据相减 取有效数据(销量大于0) 关联不上的数据当做新增
    */
    def miyaCaculate(spark: SparkSession, sourcePath: String, lastMonthPath: String): DataFrame = {

      //(商品最后一个销量 - 第一个销量) + (商品最后一个销量 - 第一个销量)/有销量天数 * n天（凑18天）


      //================1.当月集合======================================
      var uy = spark.read.orc(s"${sourcePath}").toJSON.rdd.map(x => {
        val nObject = JSON.parseObject(x)

        //提取当月集合中：add_to_field中最后一次采集的commentCount来计算销量
        val Base_Info = nObject.getOrDefault("Base_Info", "-1").toString
        val add = nObject.getJSONArray("add_to_field")
        var commentCount = add.getJSONObject(add.size() -1).get("commentCount").toString
        var commentCount_first = add.getJSONObject(0).get("commentCount").toString
        var commentator = add.getJSONObject(add.size() -1).get("commentator").toString
        //价格
        var priceText = add.getJSONObject(add.size() -1).get("priceText").toString.toDouble
        if(commentCount.equals("-1")){
          commentCount="0"
        }


        var penghuaName = ""
        if (!Base_Info.equals("-1")) {
          val base = JSON.parseObject(Base_Info)
          penghuaName = base.getOrDefault("分类", "").toString
        }
        nObject.put("penghuaName", penghuaName)
        nObject.put("commentCount", commentCount.toInt)
        nObject.put("commentCount_first", commentCount_first.toInt)
        nObject.put("add_size", add.size())
        nObject.put("commentator", commentator)
        nObject.put("priceText_collect", priceText) //原价
        nObject.put("priceText", priceText)

        nObject.toString
      })
      spark.read.json(uy).registerTempTable("month_66")


      spark.sql(
        """
          |select
          |a.*,
          |cast((a.commentCount - a.commentCount_first) as bigint) sellCount_mid
          |from
          |month_66 a
        """.stripMargin
      ).registerTempTable("all_mid")

      spark.sql(
        """
          |select
          |*,
          |cast(sellCount_mid as bigint) sellCount
          |from
          |all_mid
          |""".stripMargin
      ).registerTempTable("all")

//      gy_no.registerTempTable("all_no_gy")
      //新增的保存,但不计入计算销量
       // .repartition(4).write.orc(newPath)


      var frame33 = spark.sql(
        """
          |select
          |*,
          |cast(sellCount * priceText as decimal(20,2)) salesAmount
          |from all
        """.stripMargin
      )
      frame33.where("good_id='2571793'").show()
      frame33
    }



  /** *
    * 关联分类
    *
    * @param spark
    * @param frame
    * @return 关联好分类的所有商品信息
    */
  def miyaCate(spark: SparkSession, frame: DataFrame): DataFrame = {
    frame.registerTempTable("JAN_101")
    spark.read.json(subPath).dropDuplicates("subCategoryId").registerTempTable("FEB_2")
    val miya_3 = spark.sqlContext.sql(
      """
        |select
        |case
        |when subCategoryId ='1083'  then (case when title rlike '方便面' or title like '泡面' or title like '拉面' or title like '包拌面' or title rlike '汤面' then '10021'
        |                                       when title rlike '油'    and title not rlike '酱油' then '10021'
        |                                       when title rlike '调料'  or title rlike '调味' then '10021'
        |                                       when title rlike '杂粮'     then '10021' else '10021' end)
        |when subCategoryId ='945'   then (case when title rlike '凉席'  or title rlike '席' then '10022' else '10022' end)
        |when subCategoryId ='7915'  then (case when title rlike '凉席'  or title rlike '席' then '10022' else '10022' end)
        |when subCategoryId ='948'   then (case when title rlike '男'       then '10011' else '10011' end)
        |when subCategoryId ='2408'  then (case when title rlike '蒸汽眼罩' or title rlike '蒸汽' and title rlike '眼罩'  then '10025' else '10019' end)
        |when subCategoryId ='17513' then (case when title rlike '蒸汽眼罩' or title rlike '蒸汽' and title rlike '眼罩'  then '10025' else '10022' end)
        |when subCategoryId ='632'   then (case when title rlike '牙刷'    or title rlike '刷头' then
        |                                 (case when  title rlike '振动'   or title rlike '电动' or title rlike '震动'  then '10014'
        |                                       when title not rlike '振动' and title not rlike '电动' and title not rlike '震动' then '10019' else '10019' end) else '10019' end)
        |when subCategoryId ='879'   then (case when title rlike '牙刷' or title rlike '刷头' then (case when  title rlike '振动' or title rlike '电动' or title rlike '震动'  then '10014'
        |                                       when title not rlike '振动' and title not rlike '电动' and title not rlike '震动' then '10019' else '10019' end) else '10019' end)
        |when subCategoryId= '886'   then (case when title rlike '蒸汽' and title rlike '眼罩' then '10025' else '10019'  end)
        |when subCategoryId= '17440' then (case when title rlike '淋浴' and title rlike '花洒' then '10022' else '10022'  end)
        |
        |--乳液10019010301   面霜10019010302
        |when subCategoryId= '4097' then '10019'
        |when subCategoryId= '882' then  '10019'
        |--吸尘器
        |when subCategoryId= '17332' then  '10014'
        |
        |else '10099'
        |end firstCategoryId,
        |case
        |when subCategoryId ='1083'  then (case when title rlike '方便面' or title like '泡面' or title like '拉面' or title like '包拌面' or title rlike '汤面' then '1002102'
        |                                       when title rlike '油'and title not rlike '酱油' then '1002102'
        |                                       when title rlike '调料' or title rlike '调味' then '1002102'
        |                                       when title rlike '杂粮' then '1002102' else '1002102' end)
        |when subCategoryId ='945'   then (case when title rlike '凉席' or title rlike '席' then '1002205' else '1002205' end)
        |when subCategoryId ='7915'  then (case when title rlike '凉席' or title rlike '席' then '1002205' else '1002205' end)
        |when subCategoryId ='948'   then (case when title rlike '男' then '1001103' else '1001104' end)
        |when subCategoryId ='2408'  then (case when title rlike '蒸汽眼罩' or title rlike '蒸汽' and title rlike '眼罩'  then '1002510' else '1001902' end)
        |when subCategoryId ='17513' then (case when title rlike '蒸汽眼罩' or title rlike '蒸汽' and title rlike '眼罩'  then '1002510' else '1002206' end)
        |when subCategoryId ='632'   then (case when title rlike '牙刷' or title rlike '刷头' then
        |                                 (case when title rlike '振动' or title rlike '电动' or title rlike '震动'  then '1001404'
        |                                       when title not rlike '振动' and title not rlike '电动' and title not rlike '震动' then '1001907' else '1001907' end) else '1001907' end)
        |when subCategoryId ='879'   then (case when title rlike '牙刷' or title rlike '刷头' then
        |                                 (case when  title rlike '振动' or title rlike '电动' or title rlike '震动'  then '1001404'
        |                                       when title not rlike '振动' and title not rlike '电动' and title not rlike '震动' then '1001907' else '1001907' end) else '1001907' end)
        |when subCategoryId= '886'   then (case when title rlike '蒸汽' and title rlike '眼罩' then '1002510' else '1001901'  end)
        |when subCategoryId= '17440' then (case when title rlike '淋浴' and title rlike '花洒' then '1002209' else '1002206'  end)
        |
        |--乳液10019010301   面霜10019010302
        |when subCategoryId= '4097' then '1001901'
        |when subCategoryId= '882' then  '1001901'
        |--吸尘器
        |when subCategoryId= '17332' then  '1001402'
        |
        |else 'None'
        |end secondCategoryId,
        |
        |case
        |when subCategoryId ='1083'  then (case when title rlike '方便面' or title like '泡面' or title like '拉面' or title like '包拌面' or title rlike '汤面' then '100210203'
        |                                       when title rlike '油' and title not rlike '酱油' then '100210201'
        |                                       when title rlike '调料' or title rlike '调味' then '100210202'
        |                                       when title rlike '杂粮' then '100210203' else '100210299' end)
        |when subCategoryId ='945'   then (case when title rlike '凉席' or title rlike '席' then '100220501' else '100220501' end)
        |when subCategoryId ='7915'  then (case when title rlike '凉席' or title rlike '席' then '100220501' else '100220501' end)
        |when subCategoryId ='948'   then (case when title rlike '男' then '100110399' else '100110499' end)
        |when subCategoryId ='2408'  then (case when title rlike '蒸汽眼罩' or title like '蒸汽' and title rlike '眼罩'  then '100251002' else '100190299' end)
        |when subCategoryId ='17513' then (case when title rlike '蒸汽眼罩' or title like '蒸汽' and title rlike '眼罩'  then '100251002' else 'None' end)
        |when subCategoryId ='632'   then (case when title rlike '牙刷' or title rlike '刷头' then
        |                                 (case when  title rlike '振动' or title rlike '电动' or title rlike '震动'  then '100140408'
        |                                       when title not rlike '振动' and title not rlike '电动' and title not rlike '震动' then '100190703' else '100190703' end) else '100190799' end)
        |when subCategoryId ='879'   then (case when title rlike '牙刷' or title rlike '刷头'
        |                            then (case when  title rlike '振动' or title rlike '电动' or title rlike '震动'  then '100140408'
        |                                       when title not rlike '振动' and title not rlike '电动' and title not rlike '震动' then '100190703' else '100190703' end) else '100190799' end)
        |when subCategoryId= '886'   then (case when title rlike '蒸汽' and title rlike '眼罩' then '100251002' else '100190108'  end)
        |when subCategoryId= '17440' then (case when title rlike '淋浴' and title rlike '花洒' then '100220902' else 'None'  end)
        |
        |--乳液10019010301   面霜10019010302
        |when subCategoryId= '4097' then '100190103'
        |when subCategoryId= '882' then  '100190103'
        |--吸尘器
        |when subCategoryId= '17332' then  '100140204'
        |
        |else 'None'
        |end thirdCategoryId,
        |
        |case
        |when subCategoryId ='1083'  then (case when title rlike '方便面' or title like '泡面' or title like '拉面' or title like '包拌面' or title rlike '汤面' then '10021020305'
        |                                       when title rlike '油'and title not rlike '酱油' then 'None'
        |                                       when title rlike '调料' or title rlike '调味' then 'None'
        |                                       when title rlike '杂粮' then 'None' else 'None' end)
        |when subCategoryId ='945'   then (case when title rlike '凉席' or title rlike '席' then '10022050104' else '10022050103' end)
        |when subCategoryId ='7915'  then (case when title rlike '凉席' or title rlike '席' then '10022050104' else '10022050103' end)
        |when subCategoryId ='948'   then (case when title rlike '男' then 'None' else 'None' end)
        |when subCategoryId ='2408'  then (case when title rlike '蒸汽眼罩' or title like '蒸汽' and title rlike '眼罩'  then 'None' else 'None' end)
        |when subCategoryId ='17513' then (case when title rlike '蒸汽眼罩' or title like '蒸汽' and title rlike '眼罩'  then 'None' else 'None' end)
        |when subCategoryId ='632'   then (case when title rlike '牙刷' or title rlike '刷头' then
        |                                 (case when  title rlike '振动' or title rlike '电动' or title rlike '震动'  then 'None'
        |                                       when title not rlike '振动' and title not rlike '电动' and title not rlike '震动' then 'None' else 'None' end) else 'None' end)
        |when subCategoryId ='879'   then (case when title rlike '牙刷' or title rlike '刷头' then
        |                                 (case when  title rlike '振动' or title rlike '电动' or title rlike '震动'  then 'None'
        |                                       when title not rlike '振动' and title not rlike '电动' and title not rlike '震动' then 'None' else 'None' end) else 'None' end)
        |when subCategoryId= '886'   then (case when title rlike '蒸汽' and title rlike '眼罩' then 'None' else 'None'  end)
        |when subCategoryId= '17440' then 'None'
        |
        |--乳液 10019010301   面霜 10019010302
        |when subCategoryId= '4097' then (case when title rlike '乳液' then '10019010301' when title rlike '面霜' then '10019010302' else '10019010399' end)
        |when subCategoryId= '882' then  (case when title rlike '乳液' then '10019010301' when title rlike '面霜' then '10019010302' else '10019010399' end)
        |--吸尘器
        |when subCategoryId= '17332' then  (case when title rlike '卧式' then '10014020404'
                                                 when title rlike '立式' or title rlike '手持'  or  title rlike '无线' then '10014020402'
                                                 when title rlike '桶式' then '10014020405'
                                                 when title rlike '除螨' then '10014020401'
                                                 else '10014020499' end)
        |
        |else 'None'
        |end fourthCategoryId,
        |*
        |from
        |JAN_101
        |where
        |
        |   subCategoryId="1083"
        |or subCategoryId="945"
        |or subCategoryId="948"
        |or subCategoryId="2408"
        |or subCategoryId="7915"
        |or subCategoryId="632"
        |or subCategoryId="879"
        |or subCategoryId="886"
        |or subCategoryId="1097"
        |or subCategoryId="17440"
        |or subCategoryId="1386"
        |or subCategoryId="17513"
        |or subCategoryId="4097"
        |or subCategoryId="882"
        |or subCategoryId="17332"
        |
      """.stripMargin)

    miya_3.registerTempTable("FEB_3")

    val miya_4 = spark.sqlContext.sql(
      """
        |select
        |IFNULL(b.firstCategoryId,"10099") firstCategoryId,
        |IFNULL(b.secondCategoryId,"None") secondCategoryId,
        |IFNULL(b.thirdCategoryId,"None") thirdCategoryId,
        |IFNULL(b.fourthCategoryId,"None") fourthCategoryId,
        |a.*
        |from
        |JAN_101 a
        |left join
        |FEB_2 b
        |on a.subCategoryId=b.subCategoryId
        |where
        |a.subCategoryId!="1083"
        |and a.subCategoryId!="945"
        |and a.subCategoryId!="948"
        |and a.subCategoryId!="2408"
        |and a.subCategoryId!="7915"
        |and a.subCategoryId!="632"
        |and a.subCategoryId!="879"
        |and a.subCategoryId!="886"
        |and a.subCategoryId!="1097"
        |and a.subCategoryId!="1386"
        |and a.subCategoryId!="17513"
        |and a.subCategoryId!="17440"
        |and a.subCategoryId!="4097"
        |and a.subCategoryId!="882"
        |and a.subCategoryId!="17332"
      """.stripMargin)
    miya_4.registerTempTable("FEB_4")

    /**
      *
      * 两部分合并处理None的问题
      */
    var t01 = miya_3.union(miya_4)

    t01.registerTempTable("f1")

    spark.sql(
      """
        |select
        | *,
        | firstCategoryId as firstCategoryId1,
        | case when secondCategoryId = 'None' then concat(firstCategoryId,'99') else secondCategoryId end secondCategoryId1
        | from
        | f1
      """.stripMargin)
      .registerTempTable("f2")
    spark.sql(
      """
        |select
        | *,
        | case when thirdCategoryId = 'None' then concat(secondCategoryId1,'99') else  thirdCategoryId end thirdCategoryId1
        | from
        | f2
      """.stripMargin)
      .registerTempTable("f3")
    spark.sql(
      """
        |select
        | *,
        | case when fourthCategoryId = 'None' then concat(thirdCategoryId1,'99') else  fourthCategoryId end fourthCategoryId1
        | from
        | f3
      """.stripMargin)
      .registerTempTable("f4")
    var t0_1 = spark.sql(
      """
        |select
        |*
        |from
        |f4
      """.stripMargin).drop("firstCategoryId","secondCategoryId","thirdCategoryId","fourthCategoryId")
      .withColumnRenamed("firstCategoryId1","firstCategoryId")
      .withColumnRenamed("secondCategoryId1","secondCategoryId")
      .withColumnRenamed("thirdCategoryId1","thirdCategoryId")
      .withColumnRenamed("fourthCategoryId1","fourthCategoryId")

    t0_1
  }

  /** *
    * 关联地址
    *
    * @param spark
    * @param frame
    * @return 关联好地址的所有商品信息
    */
  def miyaAddress(spark: SparkSession, frame: DataFrame): DataFrame = {
    var frameAddress = frame.withColumn("name", lit("北京花旺在线商贸有限公司"))
      .withColumn("address", lit("北京市朝阳区太阳宫中路8号冠捷大厦15层1501、1502"))
      .withColumn("province", lit("北京市"))
      .withColumn("city", lit("朝阳区"))
      .withColumn("district", lit("朝阳区"))
      .withColumn("administrative_region", lit("华北地区"))
      .withColumn("city_grade", lit("1"))
      .withColumn("city_origin", lit("市辖区"))
      .withColumn("economic_division", lit("2"))
      .withColumn("district_origin", lit("朝阳区"))
      .withColumn("if_city", lit("0"))
      .withColumn("if_district", lit("2"))
      .withColumn("if_state_level_new_areas", lit("0"))
      .withColumn("poor_counties", lit("0"))
      .withColumn("regional_ID", lit("110105"))
      .withColumn("rural_demonstration_counties", lit("0"))
      .withColumn("rural_ecommerce", lit("0"))
      .withColumn("the_belt_and_road_city", lit("0"))
      .withColumn("the_belt_and_road_province", lit("0"))
      .withColumn("the_yangtze_river_economic_zone_city", lit("0"))
      .withColumn("the_yangtze_river_economic_zone_province", lit("0"))
      .withColumn("urban_agglomerations", lit("3"))
    frameAddress
  }


  /** *
    * 校验分类条数，总销量销售额，省市县等销量销售额
    *
    * @param frame 结果数据
    * @param spark
    */
  def checkMethod(frame: DataFrame, spark: SparkSession): Unit = {

    frame.registerTempTable("Mar_1")
    //mysql连接
    /*val prop = new Properties()
    prop.put("user", "root")
    prop.put("password", "123456")*/

    //统计平台的总销售额，总销量，总条数。
    val t1 = spark.sql(
      """
        |select
        |platformId,
        |sum(salesAmount) as sum_salesAmount,
        |sum(sellCount) as sum_sellCount,
        |count(1) as data_count,
        |timeStamp from
        |Mar_1
        |group by platformId,timeStamp
      """.stripMargin)
//    t1.write.mode("append").jdbc("jdbc:mysql://192.168.0.2:3306/O2Odata_View?useUnicode=true&characterEncoding=utf8",
//      "O2Odata_View.sell_sales_table", prop)

//data_checkout/2019/${month}/${platform}/sales"
    t1.repartition(1).write.option("header","true").csv(s"${strPath}/data_checkout/2019/${month}/${platform}/sales")
//    t1.repartition(1).write.json(s"${strPath}/2019/checkout/sales/${month}/${platform}.json")

    // 统计各省市县下的销售量、销售额、条数
    val t2 = spark.sql(
      """
        |select
        |province,
        |city,
        |sum(salesAmount) salesAmount,
        |sum(sellCount) sellCount,
        |count(1) dataCount,
        |district,
        |timeStamp,
        |platformId
        |from Mar_1
        |group by province,city,timeStamp,platformId,district
      """.stripMargin)
//    t2.write.mode("append").jdbc("jdbc:mysql://192.168.0.2:3306/O2Odata_View?useUnicode=true&characterEncoding=utf8",
//      "O2Odata_View.province_table", prop)

//    t2.repartition(1).write.csv(s"${strPath}/2019/checkout/province/${month}/${platform}.csv")
    t2.repartition(1).write.option("header","true").csv(s"${strPath}/data_checkout/2019/${month}/${platform}/province")

    // 统计各个分类下的销售量、销售额、条数
    val t3 = spark.sqlContext.sql(
      """
        |select
        |platformId,
        |timeStamp,
        |firstCategoryId,
        |secondCategoryId,
        |thirdCategoryId,
        |sum(sellCount) cate_sellCount,
        |sum(salesAmount) cate_salesAmount,
        |count(1) as cate_Count
        |from Mar_1
        |group by firstCategoryId,secondCategoryId,thirdCategoryId,platformId,timeStamp
      """.stripMargin)
//    t3.write.mode("append").jdbc("jdbc:mysql://192.168.0.2:3306/O2Odata_View?useUnicode=true&characterEncoding=utf8",
//      "O2Odata_View.thirdcategroy_table", prop)

//    t3.repartition(1).write.csv(s"${strPath}/2019/checkout/category/${month}/${platform}.csv")
    t3.repartition(1).write.option("header","true").csv(s"${strPath}/data_checkout/2019/${month}/${platform}/category")
  }

}

